package com.imooc.hadoop;

/*
 * @Author Wangjin
 * @Email wangjin4901@vip.qq.com
 * @Date 2019-01-08 下午 9:20
 */

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.*;
import org.apache.hadoop.io.IOUtils;
import org.apache.hadoop.util.Progressable;
import org.junit.After;
import org.junit.Before;
import org.junit.Test;

import java.io.BufferedInputStream;
import java.io.File;
import java.io.FileInputStream;
import java.io.InputStream;
import java.net.URI;
import java.text.SimpleDateFormat;
import java.util.Date;

/**
 * Hadoop HDFS Java API 操作
 */
public class HDFSApp {
    public static final String HDFS_PATH = "hdfs://192.168.33.128:8020";

    FileSystem fileSystem = null;
    Configuration configuration = null;


    @Test
    public void mkdir() throws Exception {
        fileSystem.mkdirs(new Path("/hdfsapi/test"));
    }

    /**
     * 创建文件
     * @throws Exception
     */
    @Test
    public void create() throws Exception {
        FSDataOutputStream outputStream = fileSystem.create(new Path("/hdfsapi/test/a.txt"));
        outputStream.write("hello ".getBytes());
        outputStream.flush();
        outputStream.close();
    }

    /**
     * 查看FDSH文件内容
     * @throws Exception
     */
    @Test
    public void cat() throws Exception {
        FSDataInputStream inputStream = fileSystem.open(new Path("/hdfsapi/test/a.txt"));
        IOUtils.copyBytes(inputStream, System.out, 1024);
        inputStream.close();
    }

    /**
     * 文件重命名
     * @throws Exception
     */
    @Test
    public void rename() throws Exception {
        Path oldPath = new Path("/hdfsapi/test/a.text");
        Path newPath = new Path("/hdfsapi/test/b.txt");
        fileSystem.rename(oldPath, newPath);
    }

    /**
     * 从本地上传文件到HDFS
     * @throws Exception
     */
    @Test
    public void copyFileFromLocal() throws Exception {
        Path localPath  = new Path("E:\\字幕文件批量改名.exe");
        Path hdfsPath = new Path("/hdfsapi/test/");
        fileSystem.copyFromLocalFile(localPath, hdfsPath);
    }

    /**
     * 从本地上传文件到HDFS带进度条
     * @throws Exception
     */
    @Test
    public void copyFileFromLocalWithProgress() throws Exception {
        System.out.println(new SimpleDateFormat("yyyy-MM-dd HH:mm:ss SS").format(new Date()));
        InputStream inputStream = new BufferedInputStream(new FileInputStream(new File("E:\\迅雷下载\\ideaIU-2018.3.2.exe")));
        FSDataOutputStream outputStream = fileSystem.create(new Path("/hdfsapi/test/IDEA.exe"), new Progressable() {
            @Override
            public void progress() {
                System.out.print("."); // 带进度提醒信息
            }
        });
        IOUtils.copyBytes(inputStream, outputStream, 4096);
    }

    /**
     * 下载HDFS文件
     * @throws Exception
     */
    @Test
    public void copyToLocalFile() throws Exception {
        Path localPath = new Path("E:\\");
        Path hdfsPath = new Path("/hdfsapi/test/IDEA.exe");
        fileSystem.copyToLocalFile(false, hdfsPath, localPath, true);
    }

    /**
     * 查看某个目录下的所有文件
     * @throws Exception
     */
    @Test
    public void listFiles() throws Exception {
        FileStatus[] fileStatuses = fileSystem.listStatus(new Path("/hdfsapi/test"));
        for(FileStatus  fileStatus : fileStatuses) {
            String isDir = fileStatus.isDirectory() ? "文件夹" : "文件";
            short replication =  fileStatus.getReplication();
            long len = fileStatus.getLen();
            String path = fileStatus.getPath().toString();
            System.out.println(isDir + "\t" + replication + "\t" + len + "\t" + path);
        }
    }

    /**
     * 删除文件
     * @throws Exception
     */
    @Test
    public void removeFile() throws Exception {
        fileSystem.delete(new Path("/hdfsapi/test/"), true);
    }

    @Before
    public void setUp() throws Exception {
        configuration = new Configuration();
        fileSystem = FileSystem.get(new URI(HDFS_PATH), configuration, "root");
    }

    @After
    public void tearDown() throws Exception {
        configuration = null;
        fileSystem = null;

        System.out.println("HDFSApp down");
        System.out.println(new SimpleDateFormat("yyyy-MM-dd HH:mm:ss SS").format(new Date()));

    }
}
